/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.sss.client.DataPutter;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.datatype.PackableDouble;
import org.sss.mapreduce.datatype.PackableInt;

public class GenVec {
  public static class GenVecMapper extends Mapper {
    public void map(Context context,
                    PackableInt key, MatSeed value,
                    Output<PackableInt, PackableDouble> output) throws Exception {
      int first = value.first;
      int last  = value.last;

      for (;first < last;++first) {
        output.write(
            new PackableInt(first),
            new PackableDouble(1.0));
      }
    }
  }

  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    int nSeeds = 0;
    boolean broadcast = false;
    String[] others = client.getArgs();
    BasicParser parser = new BasicParser();
    Options options = new Options();
    options.addOption("s", true, "number of seed(if less than 1, does not use seed)");
    options.addOption("b", false, "broadcast");
    int n = 1;
    try {
      CommandLine line = parser.parse(options, others);

      if (line.getArgs().length < 1) {
        System.err.println("usage: GenVec SIZE");
        System.exit(1);
      }
      if (line.hasOption("s")) {
        nSeeds = Integer.parseInt(line.getOptionValue("s"));
      }
      broadcast = line.hasOption("b");
      n = Integer.parseInt(line.getArgs()[0]);
    }
    catch(ParseException e) {
      System.err.println(e.getMessage());
      HelpFormatter f = new HelpFormatter();
      f.printHelp("GenVec", options);
      System.exit(1);
    }

    GroupID gID = GroupID.createRandom(client, broadcast, false);
    if (nSeeds > 0) {
      System.err.println("Initialize with worker nodes.");
      GroupID gIDSeed = GroupID.createRandom(client);
      DataPutter<PackableInt, MatSeed> putter = DataPutter.create(client,
          PackableInt.class, MatSeed.class, gIDSeed);
      try {
        int nidx = 0;
        int unit = (n + nSeeds - 1)/ nSeeds;
        for (int i = 0; i < nSeeds; ++i) {
          putter.append(new PackableInt(i), new MatSeed(nidx, Math.min(nidx + unit, n), n));
          nidx += unit;
        }
        assert nidx >= n;
      } finally {
        putter.close();
      }

      JobEngine engine = new JobEngine(client);
      try {
        engine.getJobBuilder(GenVecMapper.class).addInput(gIDSeed).addOutput(gID).build();
        engine.exec();

        // Remove temporary TupleGroup.
        client.removeTupleGroup(gIDSeed);
      } finally {
        engine.close();
      }
    }
    else {
      DataPutter<PackableInt, PackableDouble> putter =
        DataPutter.create(client, PackableInt.class, PackableDouble.class, gID);

      try {
        for (int i = 0; i < n; i++) {
          putter.append(new PackableInt(i), new PackableDouble(1.0));
        }
      } finally {
        putter.close();
      }
    }
    System.out.println("group ID for the vec is: "+ gID);
    System.exit(0);
  }
}
